#pragma once

#include "AbstractRequestService.h"
#include "network/model/PublisherNode.h"
#include "broker/BrokerConfig.h"
#include "base/Message.h"
#include "RequestDispatcher.h"
#include "base/ThreadPool.hpp"

namespace zeroplus::mq
{
    class SubService : public AbstractRequestService
    {
    private:
        struct SubscribeInfo
        {
            std::string consumerGroup;
            std::string consumerId;
            std::string topic;
            bool isSubscribed;
        };
        std::map<std::string, SubscribeInfo> isTopicSubscribeMap_;
        std::shared_ptr<RedisStore> redisStore_;
        std::shared_mutex rwMtx_;
        std::unique_ptr<PublisherNode> publisher_; // 发布者
        RequestDispatcher dispatcher_;
        ThreadPool threadPool_;

    public:
        SubService(std::shared_ptr<BrokerMediator> mediator)
            : AbstractRequestService(mediator),
              redisStore_(RedisStore::getInstance(BrokerConfig::getInstance().redisAddr)),
              threadPool_(8)
        {
            publisher_ = std::make_unique<PublisherNode>(BrokerConfig::getInstance().publisherAddr, ConnectType::bind);

            dispatcher_.RegisterHandler<transport::RequestCode::SUB_MESSAGE, transport::SubscribeMessageRequest, transport::SubscribeMessageResponse>(
                std::bind(&SubService::ProcessSubscribeMessageRequest, this, std::placeholders::_1));
            dispatcher_.RegisterHandler<transport::RequestCode::SEND_ACK, transport::SendAckMessage, void>(
                std::bind(&SubService::ProcessSendAckMessage, this, std::placeholders::_1));
        }

        ~SubService()
        {
            threadPool_.Shutdown(); // 关闭线程池
        }

        std::optional<transport::ReplyCommand> ProcessRequest(const transport::RequestCommand &cmd) override
        {
            return dispatcher_.Process(cmd);
        }

        std::tuple<Bytes, std::string, bool> TryReadMessageFromStorage(const SubscribeInfo &info)
        {
            auto topic = info.topic;
            auto consumerGroup = info.consumerGroup;
            auto consumerId = info.consumerId;
            if (!info.isSubscribed)
            {
                LOG_ERROR("topic=%s has not been subscribed", topic.c_str());
                return std::make_tuple(Bytes::create(0), "", false);
            }
            if (redisStore_->GetQueueSize(topic) <= 1)
            {
                // 如果队列长度为0，相当于不存在，会触发PubService重新注册topic到topic订阅表
                LOG_ERROR("topic: %s queue length is 1 and cannot be read", topic.c_str());
                return std::make_tuple(Bytes::create(0), "", false);
            }
            auto [message, messageId, status] = redisStore_->ReadFromMQ(topic, consumerGroup, consumerId);
            if (!status)
            {
                LOG_ERROR("read error");
                return std::make_tuple(Bytes::create(0), "", false);
            }
            return std::make_tuple(message, messageId, true);
        }

        void PublishHandler()
        {
            auto topicSubscribeMap = this->getIsTopicSubscribeMap();
            for (auto pair : topicSubscribeMap)
            {
                std::string topic = pair.first;
                SubscribeInfo info = pair.second;
                auto task = [this, topic, info]()
                {
                    auto [message, messageId, status] = TryReadMessageFromStorage(info);
                    if (!status)
                    {
                        return;
                    }
                    Message<Bytes> msg(messageId, message);
                    // 一发多收情况下, 发现底层默认了广播模式, 一条消息可以被多个订阅者收到
                    publisher_->Send(topic, msg.getSerializeData());
                    // LOG_INFO("发布消息: %s, size=%d", topic.c_str(), message.size());
                };
                task();
                /*if (!threadPool_.post_task(task))
                {
                    LOG_DEBUG("线程池或许满载了");
                }*/
            }
            std::this_thread::sleep_for(std::chrono::milliseconds(NetworkFrequency::pubMsgFrequency));
        }

        transport::SubscribeMessageResponse ProcessSubscribeMessageRequest(const transport::SubscribeMessageRequest &request)
        {
            std::string consumerGroup = request.consumer_group();
            std::string consumerId = request.consumer_id();
            std::string topic = request.topic();
            setIsTopicSubscribeMap(consumerGroup, consumerId, topic, true);
            // 响应客户端
            transport::SubscribeMessageResponse response;
            response.set_code(transport::ResponseCode::SUCCESS);
            return response;
        }

        void ProcessSendAckMessage(const transport::SendAckMessage &message)
        {
            switch (message.status())
            {
            case transport::ConsumeStatus::CONSUME_SUCCESS:
            {
                redisStore_->AckMessage(message.topic(), message.message_id(), message.consumer_group());
                break;
            }
            case transport::ConsumeStatus::RECONSUME_LATER:
            {
                redisStore_->Reconsume(message.topic(), message.consumer_group(), message.consumer_id(), message.message_id());
                break;
            }
            default:
            {
                break;
            }
            }
        }

        void InternalReceive(const ServiceType &sender, const Bytes &data) override
        {
            std::string topic = data.to_string();
            setIsTopicSubscribeMap(topic, false);
        }

        std::map<std::string, SubscribeInfo> getIsTopicSubscribeMap()
        {
            std::shared_lock<std::shared_mutex> lock(rwMtx_);
            return isTopicSubscribeMap_;
        }

        void setIsTopicSubscribeMap(const std::string &topic, bool isSubscribed)
        {
            std::unique_lock<std::shared_mutex> lock(rwMtx_);
            SubscribeInfo info;
            info.isSubscribed = isSubscribed;
            if (isTopicSubscribeMap_.find(topic) == isTopicSubscribeMap_.end())
            {
                isTopicSubscribeMap_[topic] = info;
            }
        }

        void setIsTopicSubscribeMap(const std::string &consumerGroup, const std::string &consumerId, const std::string &topic, bool isSubscribe)
        {
            std::unique_lock<std::shared_mutex> lock(rwMtx_);
            SubscribeInfo info;
            info.consumerGroup = consumerGroup;
            info.consumerId = consumerId;
            info.topic = topic;
            info.isSubscribed = isSubscribe;
            if (isTopicSubscribeMap_.find(topic) == isTopicSubscribeMap_.end())
            {
                isTopicSubscribeMap_[topic] = info;
            }
            isTopicSubscribeMap_[topic] = info;
        }

        zmqpp::socket_t *GetSocket()
        {
            return publisher_->GetSocket();
        }
    };
}
